package pri.hillchen.std.concurrent.chapter5;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by hillchen on 2018/2/7.
 */
public class BoundedBufferDemo {
    static class Product{
        private int pid;
        private String pname;

        public Product(int pid, String pname) {
            this.pid = pid;
            this.pname = pname;
        }

        @Override
        public String toString() {
            return "pid:"+pid+",pname:"+pname;
        }
    }
    private static Product create(AtomicInteger atci,int total) throws InterruptedException {
        int pid = atci.addAndGet(1);
        if(pid > total){
            return null;
        }
        return new Product(pid,"Product-"+pid);
    }
    private static Product createEndProduct() {
        return new Product(-1,"Product-end");
    }
    private static void printlnProduct(Product product)throws InterruptedException {
        System.out.println(product);
    }

    static class Producer implements Runnable{
        private final BlockingQueue<Product> products;
        private final AtomicInteger atci;
        private final int total;
        private final CountDownLatch startLatch;
        private final CountDownLatch endLatch;
        private final AtomicBoolean workFlag;

        public Producer(BlockingQueue<Product> products, AtomicInteger atci, int total,CountDownLatch startLatch,CountDownLatch endLatch,AtomicBoolean workFlag) {
            this.products = products;
            this.atci = atci;
            this.total = total;
            this.endLatch = endLatch;
            this.startLatch = startLatch;
            this.workFlag = workFlag;
        }

        public void run() {
            try {
                startLatch.await();
                while (workFlag.get()){
                    Product product = create(atci,total);
                    if(product!=null){
                        products.put(product);
                    }else{
                        workFlag.set(false);
                    }
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }finally {
                endLatch.countDown();
            }
        }
    }
    static class Consumer  implements Runnable{
        private final BlockingQueue<Product> products;
        private final CountDownLatch startLatch;
        private final AtomicBoolean workFlag;

        public Consumer(BlockingQueue<Product> products,CountDownLatch startLatch,AtomicBoolean workFlag) {
            this.products = products;
            this.startLatch = startLatch;
            this.workFlag = workFlag;
        }

        public void run() {
            try {
                startLatch.await();
                while(workFlag.get()){
                    Product product = products.take();
                    if(product!=null){
                        printlnProduct(product);
                    }
                }
            } catch (InterruptedException e) {
                if(workFlag.get()){
                    e.printStackTrace();
                }
            }
        }
    }

    private static void boundedBufferTest(final int queueSize,final int total,final int producerSize,final int consumerSize) throws InterruptedException, NoSuchFieldException {

        ArrayBlockingQueue<Product> products = new ArrayBlockingQueue<Product>(queueSize);
        final AtomicInteger atci = new AtomicInteger(0);

        final CountDownLatch startLatch = new CountDownLatch(1);
        final CountDownLatch producerEndLatch = new CountDownLatch(producerSize);
        final CountDownLatch consumerEndLatch = new CountDownLatch(consumerSize);

        final AtomicBoolean producerWorkFlag = new AtomicBoolean(true);
        final AtomicBoolean consumerWorkFlag = new AtomicBoolean(true);

        final ExecutorService fixedThreadPool = Executors.newFixedThreadPool(producerSize + consumerSize);

        for(int i=0;i<producerSize;i++){
            fixedThreadPool.execute(new Producer(products,atci,total,startLatch,producerEndLatch,producerWorkFlag));
        }
        for(int i=0;i<consumerSize;i++){
            fixedThreadPool.execute(new Consumer(products,startLatch,consumerWorkFlag));
        }
        long startTime = System.currentTimeMillis();
        startLatch.countDown();
        producerEndLatch.await();
        consumerWorkFlag.set(false);
        while (!products.isEmpty()){
            Thread.sleep(10);
        }
        fixedThreadPool.shutdownNow();
        long endTime = System.currentTimeMillis();
        System.out.println("---------------work time:"+(endTime -startTime));
    }

    public static void main(String[] args) throws InterruptedException, NoSuchFieldException {
        boundedBufferTest(20,10000,5,6);
    }
}
